package com.mini.grpc.server;

import com.mini.grpc.common.Metadata;
import com.mini.grpc.common.Status;
import com.mini.grpc.common.MethodDescriptor;
import com.mini.grpc.transport.ServerStream;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;

/**
 * 服务端调用实现类
 * 
 * @param <ReqT> 请求类型
 * @param <RespT> 响应类型
 * @author Mini-gRPC
 */
public class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {
    
    private static final Logger logger = Logger.getLogger(ServerCallImpl.class.getName());
    
    private final ServerStream stream;
    private final MethodDescriptor<ReqT, RespT> method;
    private final AtomicBoolean cancelled = new AtomicBoolean(false);
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private volatile boolean headersSent = false;
    
    /**
     * 构造函数
     * 
     * @param stream 服务端流
     * @param method 方法描述符
     */
    public ServerCallImpl(ServerStream stream, MethodDescriptor<ReqT, RespT> method) {
        this.stream = stream;
        this.method = method;
    }
    
    @Override
    public void request(int numMessages) {
        if (numMessages <= 0) {
            throw new IllegalArgumentException("numMessages must be positive");
        }
        if (isCancelled()) {
            return;
        }
        
        stream.request(numMessages);
    }
    
    @Override
    public void sendHeaders(Metadata headers) {
        if (headersSent) {
            throw new IllegalStateException("Headers already sent");
        }
        if (isCancelled()) {
            return;
        }
        if (closed.get()) {
            throw new IllegalStateException("Call already closed");
        }
        
        headersSent = true;
        stream.sendHeaders(headers != null ? headers : new Metadata());
    }
    
    @Override
    public void sendMessage(RespT message) {
        if (message == null) {
            throw new IllegalArgumentException("message cannot be null");
        }
        if (isCancelled()) {
            return;
        }
        if (closed.get()) {
            throw new IllegalStateException("Call already closed");
        }
        
        // 如果还没有发送头部，先发送默认头部
        if (!headersSent) {
            sendHeaders(new Metadata());
        }
        
        try {
            // 直接发送消息，让流处理序列化
            stream.sendMessage(message);
        } catch (Exception e) {
            logger.severe("Failed to send message: " + e.getMessage());
            close(Status.INTERNAL.withDescription("Failed to serialize response"), new Metadata());
        }
    }
    
    @Override
    public void close(Status status, Metadata trailers) {
        if (!closed.compareAndSet(false, true)) {
            return; // 已经关闭
        }
        
        if (isCancelled()) {
            return;
        }
        
        // 如果还没有发送头部，先发送默认头部
        if (!headersSent) {
            sendHeaders(new Metadata());
        }
        
        // 在trailers中添加状态信息
        if (trailers == null) {
            trailers = new Metadata();
        }
        
        // 添加状态码和消息
        trailers.put(Metadata.Key.of("grpc-status", Metadata.ASCII_STRING_MARSHALLER), 
                    String.valueOf(status.getCode().value()));
        if (status.getDescription() != null) {
            trailers.put(Metadata.Key.of("grpc-message", Metadata.ASCII_STRING_MARSHALLER), 
                        status.getDescription());
        }
        
        stream.close(status, trailers);
    }
    
    @Override
    public boolean isCancelled() {
        return cancelled.get();
    }
    
    @Override
    public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
        return method;
    }
    
    /**
     * 标记调用为已取消
     */
    public void cancel() {
        cancelled.set(true);
        if (!closed.get()) {
            stream.cancel();
        }
    }
    
    /**
     * 获取底层流
     * 
     * @return 服务端流
     */
    public ServerStream getStream() {
        return stream;
    }
} 